package cn.yeziji.netty.config.handler;

import cn.yeziji.forum.basic.netty.Message;
import cn.yeziji.forum.basic.netty.user.UserMessage;
import cn.yeziji.forum.common.key.MessageKey;
import cn.yeziji.forum.common.key.RabbitMQConfig;
import cn.yeziji.netty.adapter.CommunicationAdapters;
import cn.yeziji.netty.adapter.PublicChatExitAdapter;
import cn.yeziji.netty.entity.ChatUserChannelMessage;
import cn.yeziji.netty.factory.MessageHandleFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.support.spring.PropertyPreFilters;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * websocket 信息处理器
 *
 * @author gzkemays
 * @since 2022/11/07 2:52 PM
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class NettyWebsocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
  /** 存放通道和用户关联 */
  Map<ChannelHandlerContext, UserMessage> CONNECT_USER_MSG = new ConcurrentHashMap<>(16);

  /** 存储当前连接上的通道 */
  List<ChannelHandlerContext> CONNECT_CHANNELS = new CopyOnWriteArrayList<>();
  /** 公共聊天室连接 */
  List<ChatUserChannelMessage> PUBLISH_CHAT_LIST = new CopyOnWriteArrayList<>();
  /** 私聊聊天室连接 */
  Map<String, ChannelHandlerContext> PRIVATE_CHAT_MAP = new ConcurrentHashMap<>();
  /** 打卡连接 */
  List<ChannelHandlerContext> GUESTBOOK_LIST = new CopyOnWriteArrayList<>();

  /**
   * 连接通道时候触发
   *
   * @param ctx 客户端 socketChannel
   */
  @Override
  public void channelActive(ChannelHandlerContext ctx) {
    CONNECT_CHANNELS.add(ctx);
    // 将当前连接的用户发送出去用于展示总人数
    log.info("有新的连接，当前连接数为 --> {}", CONNECT_CHANNELS.size());
  }

  /**
   * 连接通道完毕时触发
   *
   * @param ctx
   */
  @Override
  public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    // 通道建立完毕时，发送统计人数
    Message message = new Message();
    message.setType(MessageKey.HEART_FLAG);
    message.setContext(String.valueOf(CONNECT_CHANNELS.size()));
    ctx.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
    //    ctx.writeAndFlush(new TextWebSocketFrame(String.valueOf(CONNECT_CHANNELS.size())));
  }

  /**
   * 通道消息事件
   *
   * @param channelHandlerContext 客户端的连接信息
   * @param textWebSocketFrame 客户端发送的数据流
   */
  @Override
  protected void channelRead0(
      ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) {
    String text = textWebSocketFrame.text();
    log.info("客户端发送的信息 -> {}", text);
    Message message = JSON.parseObject(text, Message.class);
    Integer type = message.getType();
    String context = message.getContext();
    MessageHandleFactory handleFactory = new MessageHandleFactory(type, context);
    CommunicationAdapters execute = handleFactory.execute();
    if (execute != null) {
      execute.notice(channelHandlerContext);
      if (context != null) {
        execute.dispose(channelHandlerContext);
      }
    }
  }

  /**
   * 通道关闭
   *
   * @param ctx
   * @throws Exception
   */
  @Override
  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    removeConnect(ctx);
    log.info("客户端断开连接当前连接数 --> {}", CONNECT_CHANNELS.size());
  }

  /**
   * 通道关闭异常
   *
   * @param ctx
   * @param cause
   * @throws Exception
   */
  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    removeConnect(ctx);
    log.info("客户端连接发生异常 --> {}", CONNECT_CHANNELS.size());
  }

  /**
   * 移除链接
   *
   * @param ctx
   */
  private void removeConnect(ChannelHandlerContext ctx) {
    // 通知聊天室有用户离开了
    PublicChatExitAdapter publicChatExitAdapter =
        new PublicChatExitAdapter(MessageKey.PUBLIC_CHAT_EXIT_FLAG, null);
    publicChatExitAdapter.notice(ctx);
    // 通知留言板不再向离开用户发送消息
    GUESTBOOK_LIST.removeIf(guestBookConnect -> guestBookConnect.equals(ctx));
    // 判断是否成功链接过
    if (CONNECT_USER_MSG.containsKey(ctx)) {
      // 删除私有聊天室的连接
      PRIVATE_CHAT_MAP.remove(CONNECT_USER_MSG.get(ctx).getUsername());
      // 清空有关连接信息
      CONNECT_USER_MSG.remove(ctx);
    }
    CONNECT_CHANNELS.remove(ctx);
  }

  /**
   * 处理消息队列，用于处理服务与通知之间的通信
   *
   * <p>实时通信按理来说使用 RPC 是最好的解决方案，但是为了减少耦合度，netty 与服务间的通信采用的 RabbitMQ 来做解耦。
   *
   * @param message 消息
   */
  @RabbitListener(queuesToDeclare = @Queue(RabbitMQConfig.NOTICE_MSG))
  public void noticeServiceMessage(Message message) {
    System.out.println("接收 service 服务消费 -> " + message);
    List<String> targets = message.getTargets();
    if (!targets.isEmpty()) {
      // 传输消息时排除 targets 字段
      PropertyPreFilters.MySimplePropertyPreFilter exclude = new PropertyPreFilters().addFilter();
      exclude.addExcludes("targets");
      for (String target : targets) {
        // 获取该用户所有未处理的通知
        for (Map.Entry<ChannelHandlerContext, UserMessage> entry : CONNECT_USER_MSG.entrySet()) {
          UserMessage value = entry.getValue();
          if (value.getUsername().equals(target)) {
            // 找到对应的链接发送通知
            ChannelHandlerContext ctx = entry.getKey();
            message.buildNoticeTime(new Date());
            // 发送了就意味着消费完毕
            ctx.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(message, exclude)));
            break;
          }
        }
      }
    }
  }

  @RabbitListener(queuesToDeclare = @Queue(RabbitMQConfig.PRIVATE_CHAT_MSG))
  public void noticePrivateChat(Message message) {
    for (String target : message.getTargets()) {
      for (Map.Entry<ChannelHandlerContext, UserMessage> entry : CONNECT_USER_MSG.entrySet()) {
        UserMessage value = entry.getValue();
        if (value.getUsername().equals(target)) {
          // 找到对应的链接发送通知
          ChannelHandlerContext ctx = entry.getKey();
          // 发送了就意味着消费完毕
          ctx.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
          break;
        }
      }
    }
    log.info("private chat msg {}", message);
  }

  public Map<ChannelHandlerContext, UserMessage> getConnectUserMsg() {
    return CONNECT_USER_MSG;
  }

  public List<ChannelHandlerContext> getConnectChannels() {
    return CONNECT_CHANNELS;
  }

  public List<ChatUserChannelMessage> getPublicChatList() {
    return PUBLISH_CHAT_LIST;
  }

  public Map<String, ChannelHandlerContext> getPrivateMap() {
    return PRIVATE_CHAT_MAP;
  }

  public List<ChannelHandlerContext> getGuestbookList() {
    return GUESTBOOK_LIST;
  }
}
